#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2023/11/26 20:15
# @Author  : 新猿异码
# @Site    : https://gitee.com/chshao/autotest
# @CSDN    : https://blog.csdn.net/m0_37576542?type=blog
# @File    : TestProject.py
# @Desc    : 模块功能描述
# @Software: PyCharm
import os
import re
import time
import datetime
from autotest.kernel.modle import CException
from autotest.kernel.modle.Variable import Var
from autotest.kernel.modle.HookImpl import HookImpl
from autotest.kernel.logger import AutoLogging
from autotest.kernel.logger.AutoLogging import Log
from autotest.kernel.config.ConfReader import ConfReader
from autotest.kernel.modle.Decorator import asynch
from autotest.kernel.runner.TestSuite import TestSuite
from autotest.kernel.runner.TestRunner import TestRunner


class TestProject(object):

    def __init__(self, conf_file, **kwargs):
        self.conf_file = conf_file
        # 加载配置文件
        Var.Config = ConfReader.read(self.conf_file)
        if kwargs.get("testcase"):
            Var.Config["testcases"] = [kwargs["testcase"]]
        # 设置工程根路径、任务编号、开始时间
        Var.Project.RootPath = os.path.split(conf_file)[0]
        Var.Project.TaskId = datetime.datetime.now().strftime('%Y%m%d_%H%M%S_%f')
        Var.Project.StartTime = time.strftime('%Y-%m-%d %H:%M:%S')
        # 设置报告生成路径
        report_root_path = Var.Config.get("report", {}).get("path", "")
        if not os.path.exists(report_root_path):
            report_root_path = os.path.split(conf_file)[0]
        Var.Project.ReportPath = os.path.join(report_root_path, "report", Var.Project.TaskId)
        # 初始化日志打印模块-工程级别日志文件
        AutoLogging.addProjectLog()
        Log.info("======================Start Project======================")

    def setup(self):
        # 搜集所有待执行的用例信息
        self.collect_testcases()
        # 将用例初始信息添加至本地.db文件中
        self.write_task_setup_msg_to_db()

    def run(self):
        TestRunner.run(TestSuite.Suite)

    def teardown(self):
        Log.info("=======================End Project=======================")
        Var.Project.EndTime = time.strftime('%Y-%m-%d %H:%M:%S')
        Var.Project.CostTime = datetime.datetime.strptime(Var.Project.EndTime, '%Y-%m-%d %H:%M:%S').timestamp() - \
                               datetime.datetime.strptime(Var.Project.EndTime, '%Y-%m-%d %H:%M:%S').timestamp()
        Log.info(f"报告路径：{os.path.join(Var.Project.ReportPath, 'report.html')}")
        Log.info(f"开始时间：{Var.Project.StartTime}")
        Log.info(f"结束时间：{Var.Project.EndTime}")
        Log.info(f"用例总数：{Var.Project.Total}个")
        Log.info(f"通过用例：{Var.Project.Passed}个")
        Log.info(f"失败用例：{Var.Project.Failed}个")
        Log.info(f"阻塞用例：{Var.Project.Block}个")
        Log.info(f"未执行用例：{Var.Project.Total - Var.Project.Passed - Var.Project.Failed - Var.Project.Block}个")
        rate = 0
        if Var.Project.Total > 0:
            rate = Var.Project.Passed / Var.Project.Total
        if rate != 1:
            Log.error(f"通过率：{round(100 * rate, 2)}%")
        else:
            Log.info(f"通过率：100%")
        # 移除日志打印模块-工程级别日志文件
        AutoLogging.removeProjectLog()
        # 更新任务信息到本地数据库
        self.update_task_msg_to_db()
        # 生成测试报告
        from autotest.kernel.report import TestReport
        TestReport.load_data_from_db(Var.Project.TaskId, Var.Config.get("report", {}).get("title", "自动化测试报告"))
        TestReport.make_report(Var.Project.ReportPath)

        # 调用任务数据入库钩子函数，二次开发拓展使用
        HookImpl.write_task_database(Var.Project)

        # 将测试结果发送邮件、企业微信消息
        subject = Var.Config.get("report", {}).get("title", "自动化测试报告")
        attach_files = [os.path.join(Var.Project.ReportPath, 'report.html')]
        from autotest.utils.notify.emails import Email
        from autotest.utils.notify.wecom import Wecom
        if Var.Config.get("notify", {}).get("email", {}).get("enabled", False):
            email = Email()
            email.smtp_server = Var.Config["notify"]["email"]["server"]
            email.smtp_port = Var.Config["notify"]["email"]["port"]
            email.email_user = Var.Config["notify"]["email"]["username"]
            email.email_pwd = Var.Config["notify"]["email"]["password"]
            if email.login() == 0:
                content = f"用例总数：{Var.Project.Total}个\n" \
                          f"通过用例：{Var.Project.Passed}个\n" \
                          f"失败用例：{Var.Project.Failed}个\n" \
                          f"阻塞用例：{Var.Project.Block}个\n" \
                          f"通过率：{round(100 * rate, 2)}%\n" \
                          f"执行耗时：{Var.Project.CostTime//60}分{Var.Project.CostTime%60}秒"
                email.send_message(subject, content, attach_files)
        if Var.Config.get("notify", {}).get("wecom", {}).get("enabled", False):
            wecom = Wecom()
            wecom.webhook = Var.Config["notify"]["wecom"]["webhook"]
            content = f"用例总数：<font color='comment'>{Var.Project.Total}个</font>\n" \
                      f"通过用例：<font color='comment'>{Var.Project.Passed}个</font>\n" \
                      f"失败用例：<font color='comment'>{Var.Project.Failed}个</font>\n" \
                      f"阻塞用例：<font color='comment'>{Var.Project.Block}个</font>\n" \
                      f"通过率：<font color='comment'>{round(100 * rate, 2)}%</font>\n" \
                      f"执行耗时：<font color='comment'>{Var.Project.CostTime//60}分{Var.Project.CostTime%60}秒</font>"
            wecom.send_message(subject, content, attach_files)

    # 搜集所有需要执行的用例
    def collect_testcases(self):
        for testsuite in Var.Config["testcases"]:
            if testsuite.count(":") != 2:
                raise CException.TestCaseConfigError("format is wrong, the num of ':' is not two!")
            suite_dir, case_class, case_method = testsuite.split(":")
            suite_path = os.path.join(Var.Project.RootPath, os.sep.join(suite_dir.split('/')))
            if suite_path.endswith(".py"):  # 配置为testcase/xxx/xxx.py:xxx:xxx形式
                self.add_testcase_into_suite(suite_dir, suite_path, case_class, case_method)
            elif suite_path.endswith("*"):  # 配置为testcase/xxx/*:xxx:xxx形式
                for root_path, dir_name, file_list in os.walk(os.path.split(suite_path)[0]):
                    for file_name in file_list:
                        if not file_name.endswith(".py") or file_name.startswith("__init__"):
                            continue
                        suite_file = os.path.join(root_path, file_name)
                        suite_dir_dot = suite_file.replace(Var.Project.RootPath, '').replace(os.sep, '/')
                        self.add_testcase_into_suite(suite_dir_dot, suite_file, case_class, case_method)
            elif os.path.exists(suite_path):  # 配置为testcase/xxx:xxx:xxx形式
                for root_path, dir_name, file_list in os.walk(suite_path):
                    for file_name in file_list:
                        if not file_name.endswith(".py") or file_name.startswith("__init__"):
                            continue
                        suite_file = os.path.join(root_path, file_name)
                        suite_dir_dot = suite_file.replace(Var.Project.RootPath, '').replace(os.sep, '/')
                        self.add_testcase_into_suite(suite_dir_dot, suite_file, case_class, case_method)
            else:  # 非以上三种格式直接抛异常
                raise CException.TestCaseConfigError("the format of '{}' is wrong!".format(suite_path))
        # 记录用例总数
        Var.Project.Total = len(TestSuite.Suite)
        Log.info("===================Collect {} TestCases===================".format(len(TestSuite.Suite)))

    def add_testcase_into_suite(self, suite_file_dir, suite_file_path, case_class, case_method):
        """
        查找测试用例方法并添加到测试套
        :param suite_file_dir: py文件相对路径
        :param suite_file_path: py文件绝对路径
        :param case_class: 类名过滤条件
        :param case_method: 方法名过滤条件
        :return:
        """
        suite_path_import = suite_file_dir.replace('/', '.').strip('.py')
        module_name = suite_path_import.split('.')[-1]
        module_loader = __import__(suite_path_import, fromlist=[module_name])
        class_name_list = self.get_class_name_from_file(suite_file_path, case_class)
        for class_name in class_name_list:
            class_loader = getattr(module_loader, class_name)
            for method_name in self.get_method_name_from_class(class_loader, case_method):
                TestSuite.add_test(".".join([suite_path_import, class_name, method_name]))

    def get_class_name_from_file(self, py_file, class_match='*'):
        """
        获取文件中符合条件的所有类名
        :param py_file: py文件绝对路径
        :param class_match: 类名过滤条件
        :return:
        """
        if not os.path.exists(py_file):
            raise FileNotFoundError("has no this file '{}'".format(py_file))
        with open(py_file, 'r', encoding="utf-8") as py_file_obj:
            file_content = py_file_obj.read()
        if len(class_match) == class_match.count("*"):
            pattern = r"class\s+([A-Za-z][A-Za-z0-9_]*)"
        else:
            pattern = r"class\s+({}[A-Za-z0-9_]*)".format(class_match.strip("*"))
        class_name_list = re.findall(pattern, file_content)
        return class_name_list

    def get_method_name_from_class(self, class_loader, method_match):
        # 先获取所有符合匹配规则的测试用例
        method_names = []
        if len(method_match) == method_match.count("*"):
            for method_name in dir(class_loader):
                if method_name in Var.Suite:
                    # 处理测试用例的参数化
                    self.__handle_params_datas(Var.Suite[method_name].get("datas"), method_name, method_names)
        else:
            for method_name in dir(class_loader):
                if method_name in Var.Suite and method_match.strip("*") in method_name:
                    # 处理测试用例的参数化
                    self.__handle_params_datas(Var.Suite[method_name].get("datas"), method_name, method_names)
        # 过滤marker标记的测试用例
        marker_enabled = Var.Config.get('marker', {}).get('enabled', False)
        markers = Var.Config.get('marker', {}).get('markers', [])
        if marker_enabled is False:
            return method_names
        return [method for method in method_names if Var.Suite.get(method.split("[")[0], {}).get('marker') in markers]

    def __handle_params_datas(self, datas, method_name, method_names):
        # 处理测试用例的参数化
        if datas is not None:
            if not (isinstance(datas, list) or callable(datas)):
                raise TypeError(f"{method_name} is not a list or function")
            if callable(datas):
                datas = datas()
            for data in datas:
                method_names.append(method_name + "[%s]" % str(data))
        else:
            method_names.append(method_name)

    def write_task_setup_msg_to_db(self):
        from autotest.kernel.modle.TestModel import Task, TestCase
        Task.create_table().execute()
        TestCase.create_table().execute()
        Task.insert(task_id=Var.Project.TaskId, start_time=Var.Project.StartTime, total=Var.Project.Total).execute()

    def update_task_msg_to_db(self):
        from autotest.kernel.modle.TestModel import Task
        not_run = Var.Project.Total - Var.Project.Passed - Var.Project.Failed - Var.Project.Block
        pass_rate = 0
        if Var.Project.Total > 0:
            pass_rate = round(100 * Var.Project.Passed / Var.Project.Total, 2)
        Task.update(end_time=time.strftime('%Y-%m-%d %H:%M:%S'),
                    passed=Var.Project.Passed,
                    failed=Var.Project.Failed,
                    block=Var.Project.Block,
                    not_run=not_run,
                    tester=Var.Config.get("report", {}).get("tester", ""),
                    pass_rate=pass_rate
                    ).where(task_id=Var.Project.TaskId).execute()


if __name__ == '__main__':
    project = TestProject(r"D:\workspace\AutoTest-Python\config.yaml")
    project.setup()
    project.run()
    project.teardown()
